package com.xrui.hbase.codec.impl;

import com.google.common.base.Preconditions;
import com.xrui.hbase.codec.CellDecoder;
import com.xrui.hbase.codec.CellDecoderFactory;
import com.xrui.hbase.impl.BoundColumnReaderSpec;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.spec.CellSpec;

import java.io.IOException;

/**
 * Factory for HBase cell decoders using GenericCellDecoder to handle record-based schemas.
 */
public final class DefaultCellDecoderFactory implements CellDecoderFactory {
    /**
     * Singleton instance.
     */
    private static final DefaultCellDecoderFactory SINGLETON = new DefaultCellDecoderFactory();

    /**
     * Singleton constructor.
     */
    private DefaultCellDecoderFactory() {
    }

    /**
     * @return an instance of the DefaultCellDecoderFactory.
     */
    public static DefaultCellDecoderFactory get() {
        return SINGLETON;
    }

    /**
     * Creates a new HBase cell encoder.
     *
     * @param cellSpec Specification of the cell encoding.
     * @return a new HBase cell encoder.
     * @throws IOException on I/O error.
     */
    public <T> CellDecoder<T> create(CellSpec cellSpec) throws IOException {
        Preconditions.checkNotNull(cellSpec);
        switch (cellSpec.getCellSchema().getType()) {
            case AVRO:
                return new GenericAvroCellDecoder<>(cellSpec);
            case RAW_BYTES:
                return new RawBytesCellDecoder(cellSpec);
            case PROTOBUF:
                return new ProtobufCellDecoder<>(cellSpec);
            case CUSTOM:
                return new CustomCellDecoder<>(cellSpec);
            case CDFSTORE:
                return new CustomCellDecoder<>(cellSpec);
            default:
                throw new InternalError("Unhandled cell encoding: " + cellSpec.getCellSchema());
        }
    }

    public <T> CellDecoder<T> create(TableLayout layout, BoundColumnReaderSpec spec)
        throws IOException {
        Preconditions.checkNotNull(layout);
        Preconditions.checkNotNull(spec);
        switch (spec.getColumnReaderSpec().getEncoding()) {
            case RAW_BYTES:
                return new RawBytesCellDecoder<>(spec);
            case AVRO:
                return new GenericAvroCellDecoder<>(layout, spec);
            case PROTOBUF:
                return new ProtobufCellDecoder<>(layout, spec);
            case CUSTOM:
                return new CustomCellDecoder<>(layout, spec);
            default:
                throw new InternalError(
                    "Unhandled cell encoding in reader spec: " + spec.getColumnReaderSpec());
        }
    }
}
